#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <sqlite3.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/types.h>
#include <time.h>

#include "MQTTClient.h"
#include "cJSON.h"
#include "shmem.h"

#define STM32 2
#define MODBUS 1
#define CONFIGPATH "../config.json"
#define CLIENTID "llyreport"
#define UPTOPIC "/app/data/up"
#define DOWNOPTIC "/app/data/down"
#define SHMID "shm"
#define MSGID "msg"
#define QOS 1

//共享内存中存储的数据
union val_t {
    int b_val;    //bool类型存储空间1
    int i_val;    //整形值存储空间2
    float f_val;  //浮点值存储空间3
};

struct std_node {
    int key;              //唯一键值
    int type;             //数据点类型
    int dev_type;         //数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val;  //变化上报后需要更新旧值
    union val_t new_val;  //从共享内存取出最新数据，放到new_val中
    int ret;              //默认为-1，采集成功后设置为0，采集失败再置-1
};

//消息队列发送的数据
struct msgbuf {
    long type;
    int key;
    bool status;
};

//配置config
cJSON *config;
//配置shm
struct shm_param shm;
//mqtt连接客户端
MQTTClient client;
//mqtt中用到的一个变量
volatile MQTTClient_deliveryToken deliveredtoken;
//延迟秒数
int period;
//设备个数
int datasize;
//数据开始地址
struct std_node *datap;
//消息队列id
int msgid;
//数据库句柄
sqlite3 *db;
//阈值
double Co2;
double Depth;

//初始化mqtt连接
int mymqtt_init();
//初始化共享内存
int myshm_init();
//上传发布订阅
void msg_publish(char *p);
//接受回调函数
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
//消息成功发送回调函数
void delivered(void *context, MQTTClient_deliveryToken dt);
//连接丢失回调函数
void connlost(void *context, char *cause);
//全部上传
void sendAll();
//修改模式
int changeType(int type, int period);
//变化上报
void *varySend(void *arg);
//把当前节点中的数据加到data数组中去
void vaddData(struct std_node *p, cJSON *data);
//周期上报
void cycleReport(int sig);
//初始化共享内存中的数据
void data_init();
//初始化消息队列
int queue_init();
//设备控制往消息队列里面发信息
void mycontrol(int key, int val);
//数据库初始化
int db_init();
//数据库存储
void *dbsave(void *arg);
//查询历史数据
void selectHistory(int key, long start, long end);
//阈值控制线程
void *thresholdFunc(void *arg);

int main(int argc, char const *argv[]) {
    //初始化mqtt和共享内存
    if (mymqtt_init() < 0) {
        printf("mqtt init err\n");
        return -1;
    }
    if (myshm_init() < 0) {
        printf("shm init err\n");
        return -1;
    }
    data_init();

    if (queue_init() < 0) {
        printf("queue init err\n");
        return -1;
    }

    if (db_init() < 0) {
        printf("db init err\n");
        return -1;
    }
    //测试一下发送消息
    // msg_publish("hello");
    //初始化阈值
    cJSON *Threshold = cJSON_GetObjectItem(config, "Threshold");
    Co2 = cJSON_GetObjectItem(Threshold, "Co2")->valuedouble;
    Depth = cJSON_GetObjectItem(Threshold, "Depth")->valuedouble;
    //开启阈值控制线程
    pthread_t thtid;
    pthread_create(&thtid, NULL, thresholdFunc, NULL);
    pthread_detach(thtid);
    //确定模式
    cJSON *report = cJSON_GetObjectItem(config, "report");
    cJSON *type = cJSON_GetObjectItem(report, "type");
    int te = type->valueint;  //0-不上报，客户端主动采集；1-变化上报，即连续2次值不相等；2-周期上报
    period = cJSON_GetObjectItem(report, "period")->valueint;
    //开启数据库存储线程
    pthread_t dbtid;
    pthread_create(&dbtid, NULL, dbsave, NULL);
    pthread_detach(dbtid);
    //配置上报模式
    if (te == 0) {
        //不上报,等客户端采集
    } else if (te == 1) {
        pthread_t tid;
        pthread_create(&tid, NULL, varySend, NULL);
        pthread_detach(tid);
        //变化上报
    } else if (te == 2) {
        signal(SIGALRM, cycleReport);
        alarm(period);
        //周期上报
    }

    char ch = '1';
    do {
        ch = getchar();
    } while (ch != 'Q' && ch != 'q');

    //回收资源
    MQTTClient_unsubscribe(client, DOWNOPTIC);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    cJSON_Delete(config);
    shm_del(&shm);
    msgctl(msgid, IPC_RMID, NULL);
    return 0;
}

//初始化mqtt连接
int mymqtt_init() {
    FILE *fp = fopen(CONFIGPATH, "r");
    if (fp == NULL) {
        perror("fopen err");
        return -1;
    }
    char buf[1024];
    int ret = fread(buf, 1, 1024, fp);
    if (ret < 0) {
        perror("fread err");
        return -1;
    }
    fclose(fp);
    // printf("%s\n", buf);
    config = cJSON_Parse(buf);
    cJSON *mqtt_server = cJSON_GetObjectItem(config, "mqtt_server");
    //mqtt配置
    cJSON *addr = cJSON_GetObjectItem(mqtt_server, "addr");
    cJSON *port = cJSON_GetObjectItem(mqtt_server, "port");
    char address[64];
    sprintf(address, "tcp://%s:%d", addr->valuestring, port->valueint);

    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    MQTTClient_create(&client, address, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    // printf("%s\n", address);
    //设置回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    //订阅主题
    MQTTClient_subscribe(client, DOWNOPTIC, QOS);
    printf("mqtt connect success!\n");
    return 0;
}

//初始化共享内存
int myshm_init() {
    datasize = cJSON_GetArraySize(cJSON_GetObjectItem(cJSON_GetObjectItem(config, "stm32"), "data")) +
               cJSON_GetArraySize(cJSON_GetObjectItem(cJSON_GetObjectItem(config, "modbus"), "data"));
    if (shm_init(&shm, SHMID, sizeof(int) + datasize * sizeof(struct std_node)) < 0) return -1;
    int *intp = (int *)shm_getaddr(&shm);
    *intp = datasize;
    datap = (struct std_node *)(intp + 1);
    printf("shm init success!size:%d\n", datasize);
}

//上传发布订阅
void msg_publish(char *p) {
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, UPTOPIC, &pubmsg, &token);
}

//消息成功发送回调函数
void delivered(void *context, MQTTClient_deliveryToken dt) {
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

//接受回调函数
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    // printf("这到了没\n");
    // printf("%s\n", (char *)message->payload);
    cJSON *msg = cJSON_Parse((char *)message->payload);
    cJSON *type = cJSON_GetObjectItem(msg, "type");
    //1-采集指令 2-控制指令 3-模式修改
    if (type->valueint == 1) {  //采集指令
        cJSON *limit = cJSON_GetObjectItem(msg, "limit");
        if (!strncmp(limit->valuestring, "all", 3)) {  //全部采集
            sendAll();
        } else if (!strncmp(limit->valuestring, "part", 4)) {  //part采集
            cJSON *keys = cJSON_GetObjectItem(msg, "keys");
            int keysize = cJSON_GetArraySize(keys);
            cJSON *send = cJSON_CreateObject();
            cJSON_AddNumberToObject(send, "type", 1);
            cJSON_AddNumberToObject(send, "result", 0);
            // cJSON_AddStringToObject(send, "msg", "控制成功");
            cJSON *data = cJSON_CreateArray();
            struct std_node *p = datap;
            for (int i = 0; i < datasize; i++) {
                for (int j = 0; j < keysize; j++) {
                    if (p->key == cJSON_GetArrayItem(keys, j)->valueint) {
                        vaddData(p, data);
                    }
                }
                p++;
            }
            cJSON_AddItemToObject(send, "data", data);
            char *sendmsg = cJSON_Print(send);
            msg_publish(sendmsg);
            //释放资源
            cJSON_Delete(send);
            free(sendmsg);
        } else {
            printf("limit->valuestring err\n");
        }
        //这边可以加part后续看时间
    } else if (type->valueint == 2) {  //控制指令往消息队列加
        cJSON *data = cJSON_GetObjectItem(msg, "data");
        mycontrol(cJSON_GetObjectItem(data, "key")->valueint, atoi(cJSON_GetObjectItem(data, "val")->valuestring));
        cJSON *send = cJSON_CreateObject();
        cJSON_AddNumberToObject(send, "type", 2);
        cJSON_AddNumberToObject(send, "result", 0);
        cJSON_AddStringToObject(send, "msg", "控制成功");
        char *sendmsg = cJSON_Print(send);
        msg_publish(sendmsg);
        //释放资源
        cJSON_Delete(send);
        free(sendmsg);
    } else if (type->valueint == 3) {  //模式修改
        cJSON *data = cJSON_GetObjectItem(msg, "data");
        cJSON *reportType = cJSON_GetObjectItem(data, "type");
        cJSON *reportPeriod = cJSON_GetObjectItem(data, "period");
        if (changeType(reportType->valueint, reportPeriod->valueint) < 0) printf("change type err\n");
    } else if (type->valueint == 4) {  //历史记录
        cJSON *data = cJSON_GetObjectItem(msg, "data");
        cJSON *limit = cJSON_GetObjectItem(data, "limit");  //时间戳范围
        selectHistory(cJSON_GetObjectItem(data, "key")->valueint, cJSON_GetArrayItem(limit, 0)->valuedouble, cJSON_GetArrayItem(limit, 1)->valuedouble);
    } else if (type->valueint == 5) {
        cJSON *data = cJSON_GetObjectItem(msg, "data");
        cJSON *Coto = cJSON_GetObjectItem(data, "Co2");
        cJSON *Dep = cJSON_GetObjectItem(data, "Depth");
        Co2 = Coto->valuedouble;
        Depth = Dep->valuedouble;
    } else {
        printf("type->valueint err\n");
    }
    cJSON_Delete(msg);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

//连接丢失回调函数
void connlost(void *context, char *cause) {
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

//全部发送一次
void sendAll() {
    int size = datasize;
    struct std_node *p = datap;
    cJSON *send = cJSON_CreateObject();
    cJSON_AddNumberToObject(send, "type", 1);
    cJSON_AddNumberToObject(send, "result", 0);
    cJSON *data = cJSON_CreateArray();
    cJSON_AddItemToObject(send, "data", data);

    for (int i = 0; i < size; i++) {
        vaddData(p, data);
        p++;
    }
    char *sendmsg = cJSON_Print(send);
    msg_publish(sendmsg);

    cJSON_Delete(send);
    free(sendmsg);
}

//修改模式
int changeType(int type, int period) {
    //配置新的JSON
    cJSON *report = cJSON_CreateObject();
    cJSON_AddNumberToObject(report, "type", type);
    cJSON_AddNumberToObject(report, "period", period);
    cJSON_ReplaceItemInObject(config, "report", report);
    char *p = cJSON_Print(config);
    //写回配置文件
    FILE *fp = fopen(CONFIGPATH, "w");
    if (fp == NULL) {
        perror("fopen err");
        return -1;
    }
    char buf[1024];
    int ret = fwrite(p, strlen(p), 1, fp);
    if (ret < 0) {
        perror("fread err");
        return -1;
    }
    //释放资源
    fclose(fp);
    free(p);
    printf("report type change!\n");
    //修改完配置文件后,发送一个成功包
    cJSON *send = cJSON_CreateObject();
    cJSON_AddNumberToObject(send, "type", 3);
    cJSON_AddNumberToObject(send, "result", 0);
    cJSON_AddStringToObject(send, "msg", "修改成功");
    char *sendmsg = cJSON_Print(send);
    msg_publish(sendmsg);

    cJSON_Delete(send);
    free(sendmsg);
    return 0;
}

//变化上报
void *varySend(void *arg) {
    int size = datasize;
    int i, flag;
    struct std_node *p;
    while (1) {
        p = datap;
        cJSON *data = cJSON_CreateArray();
        flag = 0;  //判断有无数据更新
        for (i = 0; i < size; i++) {
            if (memcmp(&p->new_val, &p->old_val, sizeof(union val_t))) {
                vaddData(p, data);
                flag = 1;
            }
            p++;
        }
        if (flag == 1) {  //有新数据上报
            cJSON *send = cJSON_CreateObject();
            cJSON_AddNumberToObject(send, "type", 1);
            cJSON_AddNumberToObject(send, "result", 0);
            cJSON_AddItemToObject(send, "data", data);
            char *sendmsg = cJSON_Print(send);
            msg_publish(sendmsg);

            cJSON_Delete(send);
            free(sendmsg);
        } else {  //没有新数据,释放资源
            cJSON_Delete(data);
        }
    }
}

//把当前节点中的数据加到data数组中去
void vaddData(struct std_node *p, cJSON *data) {
    cJSON *v = cJSON_CreateObject();
    cJSON_AddNumberToObject(v, "key", p->key);
    char buf[32] = "";  //发送的都得是字符串,存一下
    if (p->type == 0) {
        sprintf(buf, "%d", p->new_val.b_val);
        cJSON_AddStringToObject(v, "val", buf);
    } else if (p->type == 1) {
        sprintf(buf, "%d", p->new_val.i_val);
        cJSON_AddStringToObject(v, "val", buf);
    } else if (p->type == 2) {
        sprintf(buf, "%.2f", p->new_val.f_val);
        cJSON_AddStringToObject(v, "val", buf);
    } else {
        printf("p-type 有问题!:%d\n", p->type);
    }
    // printf("%d:%s\n", p->key, buf);

    cJSON_AddItemToArray(data, v);
    p->old_val = p->new_val;
}

//周期上报
void cycleReport(int sig) {
    sendAll();
    alarm(period);
}

//初始化共享内存中的数据
void data_init() {
    struct std_node *p = datap;
    cJSON *stm32 = cJSON_GetObjectItem(config, "stm32");
    cJSON *modbus = cJSON_GetObjectItem(config, "modbus");

    cJSON *stmdata = cJSON_GetObjectItem(stm32, "data");
    cJSON *modbusdata = cJSON_GetObjectItem(modbus, "data");
    int i;
    for (i = 0; i < cJSON_GetArraySize(stmdata); i++) {
        cJSON *data = cJSON_GetArrayItem(stmdata, i);
        p->key = cJSON_GetObjectItem(data, "key")->valueint;
        p->type = cJSON_GetObjectItem(data, "type")->valueint;
        p->dev_type = STM32;
        p++;
    }
    for (i = 0; i < cJSON_GetArraySize(modbusdata); i++) {
        cJSON *data = cJSON_GetArrayItem(modbusdata, i);
        p->key = cJSON_GetObjectItem(data, "key")->valueint;
        p->type = cJSON_GetObjectItem(data, "type")->valueint;
        p->dev_type = MODBUS;
        p++;
    }
}

//初始化消息队列
int queue_init() {
    key_t key;
    int ret;
    char sys_cmd[256];
    char path[256];

    sprintf(path, "/tmp/ipc/msgqueue/peer/%s", MSGID);
    sprintf(sys_cmd, "%s %s", "touch", path);
    ret = system(sys_cmd);
    UNUSED(ret);

    //创建key
    key = ftok(path, 'j');
    if (key < 0) {
        perror("queue ftok err");
        printf("error :path = %s\n", path);
        return -1;
    }

    //创建消息队列
    msgid = msgget(key, IPC_CREAT | IPC_EXCL | 0666);
    if (msgid < 0) {
        if (errno == EEXIST) {
            msgid = msgget(key, 0666);
        } else {
            perror("msgget err");
            return -1;
        }
    }
    printf("queue init success\n");
    return 0;
}

//设备控制往消息队列里面发信息
void mycontrol(int key, int val) {
    // printf("%d:%s\n", key, string);
    struct std_node *p = datap;
    struct msgbuf buf;
    for (int i = 0; i < datasize; i++) {
        if (p->key == key) {
            buf.key = key;
            buf.type = (long)p->dev_type;
            // printf("%d\n", p->dev_type);
            buf.status = val;
            printf("%d:%d\n", buf.key, buf.status);
            // if (p->type == 0) {
            //     buf.val.b_val = atoi(string);
            // } else if (p->type == 1) {
            //     buf.val.i_val == atoi(string);
            // } else if (p->type == 2) {
            //     buf.val.f_val = atof(string);
            // }
            msgsnd(msgid, &buf, sizeof(struct msgbuf) - sizeof(long), 0);
            // printf("%d:%s\n", key, string);
            break;
        }
        p++;
    }
}

// //回复信息
// void reply(int type, cJSON *data) {
//     cJSON *send = cJSON_CreateObject();
//     cJSON_AddNumberToObject(send, "type", type);
//     cJSON_AddNumberToObject(send, "result", 0);
// }

int db_init() {
    char *errmsg = NULL;
    if (sqlite3_open("./history.db", &db) != SQLITE_OK) {
        fprintf(stderr, "open user.db err %s", sqlite3_errmsg(db));
        return -1;
    }

    if (sqlite3_exec(db, "create table if not exists history(key int,val char,time long);", NULL, NULL, &errmsg) != SQLITE_OK) {
        fprintf(stderr, "create history err %s\n", errmsg);
    }

    printf("db init success\n");
}

//数据库存储线程
void *dbsave(void *arg) {
    char *errmsg = NULL;
    struct std_node *p;
    int i;                //循环变量
    time_t current_time;  //获取时间戳
    char buf[32] = "";    //存储的都是字符串,转换一下
    char sql[128] = "";   //sql语句
    while (1) {
        sleep(30);
        time(&current_time);
        p = datap;
        for (i = 0; i < datasize; i++) {
            int key = p->key;
            if (p->type == 0) {
                sprintf(buf, "%d", p->new_val.b_val);
            } else if (p->type == 1) {
                sprintf(buf, "%d", p->new_val.i_val);
            } else if (p->type == 2) {
                sprintf(buf, "%.2f", p->new_val.f_val);
            } else {
                printf("p-type 有问题!:%d\n", p->type);
            }
            sprintf(sql, "insert into history values(\"%d\",\"%s\",\"%ld\");", key, buf, current_time);
            if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK) {
                fprintf(stderr, "insert history err %s\n", errmsg);
            }
            p++;
        }
    }
}

//查询历史数据
void selectHistory(int key, long start, long end) {
    char sql[128] = "";
    char **result;
    int row = 0, column = 0;
    char *errmsg = NULL;
    int j;
    char value[32];

    cJSON *send = cJSON_CreateObject();
    cJSON_AddNumberToObject(send, "type", 4);
    cJSON_AddNumberToObject(send, "result", 0);
    cJSON_AddNumberToObject(send, "key", key);
    cJSON *data = cJSON_CreateArray();

    sprintf(sql, "select * from history where key = \"%d\" and time between \"%ld\" and \"%ld\";", key, start, end);
    if (sqlite3_get_table(db, sql, &result, &row, &column, &errmsg) != SQLITE_OK) {
        fprintf(stderr, "select err:%s", errmsg);
    }

    int k = column;  //取数据
    for (j = 0; j < row; j++) {
        cJSON *v = cJSON_CreateObject();
        k++;
        cJSON_AddStringToObject(v, "val", result[k++]);
        cJSON_AddNumberToObject(v, "time", strtol(result[k++], NULL, 10));
        cJSON_AddItemToArray(data, v);
    }
    cJSON_AddItemToObject(send, "data", data);
    char *sendmsg = cJSON_Print(send);
    msg_publish(sendmsg);

    cJSON_Delete(send);
    free(sendmsg);
}

//阈值控制线程
void *thresholdFunc(void *arg) {
    // co2;
    // Depth;
    struct std_node *p;
    int ledStatus;
    int fanStatus;
    int doorStatus;
    int beepStatus;
    float co2Val;
    float h2oVal;
    while (1) {
        p = datap;
        for (int i = 0; i < datasize; i++) {
            if (p->key == 307) {
                ledStatus = p->new_val.b_val;
            } else if (p->key == 301) {
                fanStatus = p->new_val.b_val;
            } else if (p->key == 302) {
                doorStatus = p->new_val.b_val;
            } else if (p->key == 304) {
                beepStatus = p->new_val.b_val;
            } else if (p->key == 305) {
                co2Val = p->new_val.f_val;
            } else if (p->key == 306) {
                h2oVal = p->new_val.f_val;
            }
            p++;
        }
        if (ledStatus == 0 && doorStatus == 1) {
            mycontrol(307, 1);
        }
        if (co2Val > Co2 && fanStatus == 0) {
            mycontrol(301, 1);
        }
        if (h2oVal > h2oVal && beepStatus == 0) {
            mycontrol(304, 1);
        }
    }
}
